package platform.llkz.server.common;

import io.netty.channel.*;

import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

public abstract class MyServerCommonHandler extends SimpleChannelInboundHandler<String> {
   protected boolean sentFlag;
    private Runnable counterTask;
    private AtomicLong atomicLong =new AtomicLong();
    private long pro;

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        counterTask = ()->{
          while (true){
              try{
                  Thread.sleep(500);
                  long length = atomicLong.getAndSet(0);
                  if(0 == length) continue;
                  System.out.println("数据发送速率（KB/S）:"+length);
              }catch (Exception e){
                  e.printStackTrace();
              }
          }
        };
        super.handlerAdded(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        // 启动监控线程
        new Thread(counterTask).start();
    }

    protected abstract void sendData(ChannelHandlerContext ctx);


    protected ChannelProgressivePromise getChannelProgressivePromise(ChannelHandlerContext ctx, Consumer<ChannelProgressiveFuture> consumer){
        ChannelProgressivePromise channelProgressivePromise = ctx.newProgressivePromise();
        channelProgressivePromise.addListener(new ChannelProgressiveFutureListener() {
            @Override
            public void operationProgressed(ChannelProgressiveFuture channelProgressiveFuture, long l, long l1) throws Exception {
                atomicLong.addAndGet(l-pro);
                pro = l;
            }

            @Override
            public void operationComplete(ChannelProgressiveFuture channelProgressiveFuture) throws Exception {
                sentFlag = false;
                if(channelProgressiveFuture.isSuccess()){
                    System.out.println("消息发送成功");
                    pro -= 10;
                    Optional.ofNullable(consumer).ifPresent(action -> action.accept(channelProgressiveFuture));
                }
            }
        });
        return channelProgressivePromise;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        System.out.println("server接到消息:"+s);
    }
}
